import com.fasterxml.jackson.annotation.JsonFormat;
import com.fasterxml.jackson.annotation.JsonInclude;
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerRecord;

import java.util.Date;
import java.util.Properties;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

@Slf4j
public class SuperviseService {

    public static final String SUPERVISE_TOPIC = "supervise";

    public static final Producer<String, String> PRODUCER;

    static {
        Properties KAFKA_PRODUCER_PROPS = new Properties();
        KAFKA_PRODUCER_PROPS.put("bootstrap.servers", "47.92.23.67:9092");
        KAFKA_PRODUCER_PROPS.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        KAFKA_PRODUCER_PROPS.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        PRODUCER = new KafkaProducer<>(KAFKA_PRODUCER_PROPS);
    }

    private static final ScheduledExecutorService scheduler = Executors.newSingleThreadScheduledExecutor();

    public static int COUNT = 0;

    public static void start() {
        log.info("SuperviseService started.");
        // 定时任务：每5秒执行一次
        scheduler.scheduleAtFixedRate(SuperviseService::sendSensorData, 0, 10, TimeUnit.SECONDS);
        // 添加关闭钩子
        Runtime.getRuntime().addShutdownHook(new Thread(() -> {
            shutdown();
            PRODUCER.close();
        }));
    }

    public static void shutdown() {
        scheduler.shutdown();
        try {
            if (!scheduler.awaitTermination(60, TimeUnit.SECONDS)) {
                scheduler.shutdownNow();
            }
        } catch (InterruptedException e) {
            log.error("Shutdown interrupted", e);
            scheduler.shutdownNow();
        }
        log.info("SuperviseService is shutting down.");
    }

    private static void sendSensorData() {
        try {
            Date createTime = new Date();
            SensorData sensorData = new SensorData();
            sensorData.setFlame(RaspberrySensor.flame());
            sensorData.setLight(RaspberrySensor.light());
            sensorData.setMq4(RaspberrySensor.mq4());
            sensorData.setMq7(RaspberrySensor.mq7());
//            sensorData.setDht11(RaspberrySensor.temperatureAndHumidity());
            sensorData.setInfrared(RaspberrySensor.infrared());
            sensorData.setSound(RaspberrySensor.sound());
            sensorData.setCreateTime(createTime);
            if (COUNT == 3) {
                sensorData.setDht11(RaspberrySensor.temperatureAndHumidity());
                COUNT = 0;
            }
            sendData(SUPERVISE_TOPIC, sensorData);
            COUNT += 1;
        } catch (Exception e) {
            log.error("Error sending sensor data to Kafka", e);
        }
    }


    public static void sendData(String topic, Object data) {
        String serializeData = JacksonUtil.serialize(data);
        PRODUCER.send(new ProducerRecord<>(topic, serializeData));
        log.info("Sent data to Kafka: {}", serializeData);
    }

    /**
     * 温湿度 30s一次
     * MQ4  10S一次 数字 状态
     * MQ7
     * 火焰
     * 光敏
     */
    @Data
    @JsonInclude(JsonInclude.Include.NON_NULL)
    public static class SensorData {
        private Dht11 dht11;
        private SensorValue mq4;
        private SensorValue mq7;
        private SensorValue flame;
        private SensorValue light;
        private SensorValue infrared;
        private SensorValue sound;
        @JsonFormat(pattern = "yyyy-MM-dd HH:mm:ss", timezone = "GMT+8")
        private Date createTime;

    }

    @Data
    public static class Dht11 {
        private Double temperature;
        private Double humidity;
    }

    @Data
    @AllArgsConstructor
    public static class SensorValue {
        private Integer value;
        private Boolean status;
    }

}
